package org.example.udf;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.annotation.InputGroup;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;
import org.example.data.WaterSensor;

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;

/**
 * 自定义表函数（一进多出）
 */
public class MyTableFunctionDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> strDS = env.fromElements(
                "hello flink",
                "hello world hi",
                "hello java"
        );

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //指定字段
        //Table table = tableEnv.fromDataStream(sensorDS,$("id"));
        Table strTable = tableEnv.fromDataStream(strDS, $("words"));
        tableEnv.createTemporaryView("str", strTable);

        //注册函数
        tableEnv.createTemporarySystemFunction("SplitFunction", SplitFunction.class);

        //交叉连接
        //tableEnv.sqlQuery("select words,word,length from str,lateral table(SplitFunction(words))")
        //左连接带上on true条件
        //tableEnv.sqlQuery("select words,word,length from str left join lateral table(SplitFunction(words)) on true ")
        //重命名字段
        tableEnv.sqlQuery("select words,newWord,newLength from str left join lateral table(SplitFunction(words)) as t(newWord,newLength) on true ")
                .execute()
                .print();

    }

    @FunctionHint(output = @DataTypeHint(value = "ROW<word STRING,length INT>", inputGroup = InputGroup.ANY))
    public static class SplitFunction extends TableFunction<Row> {
        //接入任何类型的输入，返回int类型输出
        public void eval(String str) {
            String[] split = str.split(" ");
            for (String s : split) {
                collect(Row.of(s, s.length()));
            }
        }
    }
}
